package ods

import java.util

import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import modes.TableObject
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer


object FromKafkaToHBase {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val consumer: FlinkKafkaConsumer[String] = SourceKafka.getSource("stage9_binlog")
    val src: DataStream[String] = env.addSource(consumer)

    val mapped: DataStream[util.ArrayList[TableObject]] = src.map((str: String) => {
      val json: JSONObject = JSON.parseObject(str)
      val database: String = json.getString("database")
      val table: String = json.getString("table")
      val typeInfo: String = json.getString("type")
      val dataInfos: JSONArray = json.getJSONArray("data")
      val objects = new util.ArrayList[TableObject]()
      if (dataInfos != null) {
        dataInfos.forEach(x =>
          objects.add(TableObject(database, table, typeInfo, x.toString))
        )
      }
      objects
    })

    mapped.addSink(new HBaseSink)

    env.execute()
  }
}
